package com.shuhe.metastore;


import com.shuhe.common.MyDateUtils;
import com.shuhe.domain.alarm.TopicMessage;
import com.shuhe.domain.alarm.TopicTable;
import com.shuhe.handler.AlarmHandler;
import com.shuhe.thread.MqttPublishThread;
import com.shuhe.thread.MqttReceiveThread;
import org.apache.log4j.Logger;
import org.eclipse.paho.mqttv5.client.*;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;

import java.sql.Connection;
import java.sql.ResultSet;
import java.util.ArrayList;

import static com.shuhe.common.CommonFunction.closeDb;
import static com.shuhe.thread.MqttPublishThread.sendMessage;

public class MqttClientV5 {

    private static Logger logger = Logger.getLogger(MqttClientV5.class);

    public static void testConnect(String brokerUrl, String client_id, String username, String password) throws Exception {
        MqttClient client = new MqttClient(brokerUrl, client_id, new MemoryPersistence());
        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setUserName(username);
        options.setPassword(password.getBytes());
        options.setCleanStart(true);
        options.setAutomaticReconnect(true); // 启用自动重连
        options.setKeepAliveInterval(60); // 设置心跳间隔（秒）
        options.setSessionExpiryInterval(0L);
        client.connect(options);
        client.disconnect();
    }

    public static void testSubsription(String brokerUrl, String client_id, String username, String password,
                                       String topic, boolean isAutoClose) throws Exception {
        MqttClient client = new MqttClient(brokerUrl, client_id, new MemoryPersistence());
        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setUserName(username);
        options.setPassword(password.getBytes());
        options.setCleanStart(true);
        options.setAutomaticReconnect(true); // 启用自动重连
        options.setKeepAliveInterval(60); // 设置心跳间隔（秒）
        options.setSessionExpiryInterval(0L);

        // 设置回调
        client.setCallback(new MqttCallback() {
            @Override
            public void disconnected(MqttDisconnectResponse disconnectResponse) {
                logger.info("与 Broker 连接断开: " + disconnectResponse.getReasonString());
                TopicMessage subjectMessage = new TopicMessage();
                subjectMessage.setTopic(topic);
                subjectMessage.setReceive_time(MyDateUtils.getNowDate());
                subjectMessage.setMessage("断开成功: " + brokerUrl);
                AlarmHandler.testSubjectMessages.add(subjectMessage);
            }

            @Override
            public void mqttErrorOccurred(MqttException exception) {
                logger.error("MQTT 错误: " + exception.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                TopicMessage subjectMessage = new TopicMessage();
                subjectMessage.setTopic(topic);
                subjectMessage.setReceive_time(MyDateUtils.getNowDate());
                subjectMessage.setMessage(new String(message.getPayload()));
                AlarmHandler.testSubjectMessages.add(subjectMessage);
                if (isAutoClose) {
                    AlarmHandler.stopTopics.add(topic);
                }
            }

            @Override
            public void deliveryComplete(IMqttToken token) {
                // 主要用于发布消息的确认，订阅方可不重点实现
            }

            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                logger.info("连接成功: " + (reconnect ? "重连" : "首次连接") + ": " + serverURI);
                TopicMessage subjectMessage = new TopicMessage();
                subjectMessage.setTopic(topic);
                subjectMessage.setReceive_time(MyDateUtils.getNowDate());
                subjectMessage.setMessage("连接成功: " + brokerUrl);
                AlarmHandler.testSubjectMessages.add(subjectMessage);
            }

            @Override
            public void authPacketArrived(int reasonCode, MqttProperties properties) {
                // 处理认证包，仅在需要特殊认证时使用
            }
        });

        int pos = -1;
        int j = 0;
        while (j < AlarmHandler.mqttClients.size()) {
            MqttClient mqttClient = AlarmHandler.mqttClients.get(j);
            if (!mqttClient.isConnected()) {
                AlarmHandler.mqttClients.remove(j);
                continue;
            }
            if (mqttClient.getClientId().equals(client_id) && mqttClient.getServerURI().equals(brokerUrl)) {
                pos = j;
                break;
            }
            j++;
        }
        if (pos == -1) {
            // 连接到 Broker
            client.connect(options);
            AlarmHandler.mqttClients.add(client);
        } else {
            client = AlarmHandler.mqttClients.get(pos);
        }

        // 订阅主题
        client.subscribe(topic, Integer.parseInt("0"));

        // 保持程序运行以持续接收消息
        int i = 0;
        while (true) {
            Thread.sleep(1000);
            i++;
            if (i > 60 * 10) {
                client.disconnect();
                break;
            }
            if (AlarmHandler.stopTopics.contains(topic)) {
                client.disconnect();
                break;
            }
        }
    }


    public static void testPublish(String brokerUrl, String client_id, String username, String password,
                                   String topic) throws Exception {
        MqttAsyncClient client = null;
        try {
            client = new MqttAsyncClient(brokerUrl, client_id, new MemoryPersistence());
            MqttConnectionOptions options = new MqttConnectionOptions();
            options.setUserName(username);
            options.setPassword(password.getBytes());
            options.setCleanStart(true);
            options.setAutomaticReconnect(true); // 启用自动重连
            options.setKeepAliveInterval(60); // 设置心跳间隔（秒）
            options.setSessionExpiryInterval(0L);

            // 连接并等待连接完成
            IMqttToken connToken = client.connect(options);
            connToken.waitForCompletion(); // 等待连接完成
            sendMessage(client, topic, "test send message");
        }finally {
            if (client != null) {
                client.disconnect();
                client.close();
            }
        }
    }



    public static void runAllSource() {
        while (true) {
            Connection appConn = null;
            ArrayList<Thread> threads = new ArrayList<Thread>();
            try {
                appConn = DBAppSql.connect();
                String sql = "select * from sys_alarminfo_collection_config_datasource";
                ResultSet rs = DBAppSql.querySql(appConn, sql);
                while (rs.next()) {
                    String source_id = rs.getString("source_id");
                    String broker_ip = rs.getString("broker_ip");
                    String port = rs.getString("port");
                    String client_id = rs.getString("client_id");
                    String username = rs.getString("username");
                    String password = rs.getString("password");
                    sql = "select * from\n" +
                            "(select subscribe_id, source_id, topic, dest_table_name, data_sample, 'receive' as " +
                            "operate_type from sys_init_alarm_app_information where status = '1'\n" +
                            "union all\n" +
                            "select id, sub_source_id, sub_topic, table_name, '', 'publish' from " +
                            "sys_mqtt_publish_config where status = '1') a where source_id = '" + source_id + "'";
                    ResultSet rs1 = DBAppSql.querySql(appConn, sql);
                    ArrayList<TopicTable> topicTables = new ArrayList<>();
                    while (rs1.next()) {
                        String operate_type = rs1.getString("operate_type");
                        String subscribe_id = rs1.getString("subscribe_id");
                        String topic = rs1.getString("topic");
                        String dest_table_name = rs1.getString("dest_table_name");
                        String data_sample = rs1.getString("data_sample");
                        sql = "SELECT TABLE_NAME\n" + "FROM information_schema.TABLES\n" + "WHERE TABLE_SCHEMA = " +
                                "DATABASE()      \n" + "and lower(table_name) = '" + dest_table_name.toLowerCase() +
                                "' ";
                        ResultSet rs2 = DBAppSql.querySql(appConn, sql);
                        TopicTable topicTable = new TopicTable();
                        topicTable.setTopic(topic);
                        topicTable.setDest_table_name(dest_table_name);
                        topicTable.setSubscribe_id(subscribe_id);
                        if (rs2.next()) {
                            topicTable.setIs_exist(true);
                        } else {
                            topicTable.setIs_exist(false);
                        }
                        if (data_sample != null && !data_sample.equals("")) {
                            topicTable.setIs_blank(false);
                        } else {
                            topicTable.setIs_blank(true);
                        }
                        topicTable.setOperateType(operate_type);
                        topicTables.add(topicTable);
                    }
                    String brokerUrl = "tcp://" + broker_ip + ":" + port;
                    ArrayList<TopicTable> receiveTopicTables = new ArrayList<>();
                    ArrayList<TopicTable> publishTopicTables = new ArrayList<>();

                    for (TopicTable topicTable : topicTables) {
                        if ("receive".equals(topicTable.getOperateType())) {
                            receiveTopicTables.add(topicTable);
                        } else if ("publish".equals(topicTable.getOperateType())) {
                            publishTopicTables.add(topicTable);
                        }
                    }
                    if (receiveTopicTables.size() > 0) {
                        Thread thread = new Thread(new MqttReceiveThread(brokerUrl, client_id, username, password,
                                receiveTopicTables, appConn));
                        thread.start();
                        threads.add(thread);
                    }
                    if (publishTopicTables.size() > 0) {
                        Thread thread = new Thread(new MqttPublishThread(brokerUrl, client_id, username, password,
                                publishTopicTables, appConn));
                        thread.start();
                        threads.add(thread);

                    }
                    Thread.sleep(1000);
                }
                for (Thread thread : threads) {
                    thread.join();
                }
                logger.info("完成一轮，重新读取配置信息");

            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                closeDb(appConn);
            }
        }

    }


    public static void main(String[] args) throws Exception {
        Thread client1Thread = new Thread(() -> {
            try {
                testSubsription("tcp://localhost:1883", "client_1", "admin", "123456", "alarms/+", false);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });


        // 启动两个线程
        client1Thread.start();

        logger.info("测试退出...");
    }
}
